package com.atguigu.cn.dataStream.sink

import java.sql.{Connection, DriverManager, PreparedStatement}

import com.atguigu.cn.dataStream.api.SensorReading
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._

/**
  * @author: yangShen
  * @Description:
  * @Date: 2020/4/15 15:44 
  */
object JdbcSinkTest {
  def main(args: Array[String]): Unit = {

    //1.environment
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)

    // 2.source
    val inputStream = environment.readTextFile("D:\\my\\myGit\\mayun\\miaohui8023\\my-flink\\flink-tutorial\\src\\main\\resources\\sensor.txt")

    // 3.transform 操作，但没有标记返回类型
    val dataStream = inputStream.map(data => {
      val dataArray = data.split(",")
      SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
    })

    //4.sink
    dataStream.addSink(new MyJdbcSink())

    environment.execute("jdbc sink test")


  }
}

//自定义数据库连接：把数据流中的数据写入到数据库
class MyJdbcSink extends RichSinkFunction[SensorReading]{
  //定义sql连接，预编译器
  var conn : Connection = _
  var insertStmt : PreparedStatement = _
  var updateStmt : PreparedStatement = _

  //初始化，创建连接和预编译语句
  override def open(parameters: Configuration): Unit = {
    super.open(parameters)
    conn = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306/test", "root", "")
    insertStmt = conn.prepareStatement("insert into sensor_reading (id, timestamp, temperature) values (?,?,?)")
    updateStmt = conn.prepareStatement("update sensor_reading set temperature = ? where id = ?")
  }

  //调用连接，执行sql
  override def invoke(sensorReading: SensorReading, context: SinkFunction.Context[_]): Unit = {
    //执行更新语句
    updateStmt.setDouble(1,sensorReading.temperature)
    updateStmt.setString(2,sensorReading.id)
    updateStmt.execute()
    if (updateStmt.getUpdateCount == 0){
      insertStmt.setString(1,sensorReading.id)
      insertStmt.setLong(2, System.currentTimeMillis())
      insertStmt.setDouble(3, sensorReading.temperature)
      insertStmt.execute()
    }
  }

  //关闭资源
  override def close(): Unit = {
    insertStmt.close()
    updateStmt.close()
    conn.close()
  }
}
